package com.htdata.iiot;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import com.alibaba.fastjson.JSONObject;
import com.htdata.iiot.api.pojo.DeviceNumber;
import com.htdata.iiot.api.pojo.StreamingData;
import com.htdata.iiot.api.pojo.StreamingMetricData;
import com.htdata.iiot.db.DbConnectionPool;
import com.htdata.iiot.db.JDBCUtils;
/**
 * 黑牡丹历史
 * @author SunMy
 *
 */
public class Main4 {
	private static DbConnectionPool pool = DbConnectionPool.getPool();
	private static String topic;
	public static void main(String[] args) {

		for(String arg:args){
			String[] split = arg.split("=");
			switch (split[0]) {
			case "topic":
				topic = split[1];
				Configs.KafkaConfig.TOPIC=topic;
				break;

			default:
				break;
			}
		}
		
		
		Map<Integer, Long> map = new HashMap<Integer,Long>();
		
		List<StreamingData> sList = new ArrayList<StreamingData>();
		SendDataToKafka sendDataToKafka = new SendDataToKafka();
		// 1.根据租户ID获取设备
		String sql1 = "SELECT tenant_id ,create_time as time, id,serial_number FROM `thing_device` "
				+ "WHERE deleted=0 AND type=0 ";
				//+ "AND create_time < '2018-03-13' ";
/*		String sql1 = "SELECT id,serial_number FROM `thing_device` "
				+ "WHERE `tenant_id` =1 AND deleted=0 AND struct_id!=0 " + "and serial_number!='SLHP-ChangZhou-001'";
*/
		
			sql1+=" and `tenant_id` in(SELECT tenant_id FROM `user` WHERE account = 'heimudan' "
					+ ") ";
		
		System.out.println(sql1);
		Connection conn = pool.getConnection();
		PreparedStatement ps = null;
		ResultSet rs = null;
		ResultSet executeQuery = null;
		PreparedStatement prepareStatement = null;
		try {
			ps = conn.prepareStatement(sql1);
			
			rs = ps.executeQuery();
			while (rs.next()) {
				int deviceId = rs.getInt("id");
				int tenant_id = rs.getInt("tenant_id");
				System.out.println(rs.getTimestamp("time"));
				long time = rs.getTimestamp("time").getTime();
				map.put(deviceId, time);
				StreamingData streamingData = new StreamingData();
				streamingData.setDid(deviceId);
				streamingData.setTid(tenant_id);
				// 2.根据设备获取测点
				String sql2 = "SELECT "
						+ "t1.id AS id,"
						+ "t2.`name` AS NAME "
						+ "FROM `thing_device_component_rel` t1,"
						+ " `thing_data_stream` t2 ,"
						+ "  thing_component t3 "
						+ "WHERE t1.`component_id` = t3.`id` "
						+ "  AND t3.deleted = 0 "
						+ "  AND t2.`component_id` = t3.`id` "
						+ "  AND t1.deleted = 0 "
						+ "  AND t1.device_id = ? "
						+ "  AND t2.deleted = 0 "
						+ "  AND t2.tenant_id = ?  ";
				prepareStatement = conn.prepareStatement(sql2);
				prepareStatement.setInt(1, deviceId);
				prepareStatement.setInt(2, tenant_id);
				executeQuery = prepareStatement.executeQuery();
				List<StreamingMetricData> mList = new ArrayList<StreamingMetricData>();
				while (executeQuery.next()) {
					String name = executeQuery.getString("name");
					long mId = executeQuery.getLong("id");
					StreamingMetricData metricData = new StreamingMetricData();
					metricData.setK(name);
					metricData.setDn(new DeviceNumber(tenant_id, deviceId, mId).toHexString());
					mList.add(metricData);
				}
				streamingData.setData(mList);
				sList.add(streamingData);
			}

		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			JDBCUtils.close(ps, rs, conn);
			JDBCUtils.close(prepareStatement, executeQuery, null);
		}
		System.out.println(sList);
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println(sList.size());
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		
		long startTime = System.currentTimeMillis();
		long i = 0;
		long mi = 0;
		
		for (StreamingData sData : sList) {
			
			System.out.println("开始发送："+sData.getDid());
			
			long st = map.get(sData.getDid());
			
			while (st<System.currentTimeMillis()) {
				st+=1000*600;

				List<StreamingMetricData> data = sData.getData();
				for (StreamingMetricData mData : data) {
					if(mData.getK().equals("status")){
						mData.setV(Math.pow(2, new Random().nextInt(3)));
					}else{
						mData.setV(Math.ceil(new Random().nextDouble() * 10000) * 0.01);
					}
					mData.setTs(st);
					mi++;
				}
				sData.setTs(st);
				sendDataToKafka.send(sData.getDid(),buildResult(sData));
				i++;
			}
			long endTime = System.currentTimeMillis();
			long useTime = endTime - startTime;
			System.out.println("写入 " + i + " 条数据,累计测点数："+mi+"，平均速率 " + (i * 1000 / useTime)+"，测点速率 " + (mi * 1000 / useTime));
		}
		System.out.println("结束。。。");

	}

	private static String buildResult(StreamingData sData) {
		return JSONObject.toJSONString(sData).toString();
	}

}
